package com.zetyun.tiger.connector.redis;


import com.fasterxml.jackson.databind.ObjectMapper;
import com.zetyun.rt.meta.annotation.ActionMeta;
import com.zetyun.rt.sdk.internal.DelegateSinkAction;
import com.zetyun.rt.sdk.model.RtEvent;
import com.zetyun.rt.sdk.operator.OperatorContext;
import org.apache.flink.streaming.connectors.redis.RedisSink;
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisSentinelConfig;

import java.util.HashSet;

@ActionMeta(
        id = "tiger.redis-sink",
        tags = {"redis"},
        category = "sink",
        name = "RtEventRedisSink",
        description = "Redis宿",
        parameterClass = RtEventRedisSinkParameter.class,
        status = ActionMeta.Status.PUBLIC
)
public class RtEventRedisSink extends DelegateSinkAction {

    @Override
    public void init(OperatorContext context) {
        super.init(context);

        ObjectMapper om = new ObjectMapper();

        RtEventRedisSinkParameter parameters = context.getParameters();

        RtEventRedisMapper mapper = new RtEventRedisMapper(
                parameters.getRc(),
                parameters.getAdditionalKey(),
                parameters.getKeyName(),
                parameters.getValueName(),
                om
        );
        FlinkJedisSentinelConfig conf =
                new FlinkJedisSentinelConfig.Builder()
                        .setMasterName(parameters.getMasterName())
                        .setSentinels(parameters.getSentinels())
                        .setPassword(parameters.getPassword())
                        .setDatabase(parameters.getDatabase())
                        .build();

        RedisSink<RtEvent> redisSinkFunction =
                new RedisSink<>(conf, mapper);

        try {
            delegate(redisSinkFunction);
        } catch (Exception e) {
            e.printStackTrace();
        }

    }
}
